package wordCount

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * 基于流计算的WordCount案例，统计每个单词的案例
 *
 * @author Liufei
 * @date 2020/5/1 12:18 下午
 */
object StreamWordCount {

  def main(args: Array[String]): Unit = {
    // 1. 初始化Flink的Streaming（流计算）上下文执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度，就是由几个线程执行
    environment.setParallelism(1)
    // 2. 导入隐式转换，建议写在这里，可以防止IDEA代码提示出错的问题
    import org.apache.flink.streaming.api.scala._
    // 3. 读取数据
    val stream = environment.socketTextStream("192.168.199.237", 8888)
    // 4. 转换计算
    val result = stream.flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0) // 分组算法：0 或者 1 代表下标。前面的DataStream[二元组]，0代表单词，1代表单词出现的次数
      .sum(1) // 聚会累加算子
    // 5. 打印到控制台
    result.print("结果是：")
    // 6. 启动流计算程序
    environment.execute("wordcount")
  }
}
